/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.example.work1.account.rocketmq;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.example.work1.account.entity.Account;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.work1.account.service.AccountService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * 生成事务监听器impl
 *
 * @author xh
 * @date 2023/03/08
 */
@Service
@Slf4j
public class ProduceTransactionListenerImpl implements TransactionListener {
    @Autowired
    AccountService accountService;
    /**
     * 本地事务方法，这个方法会在每一条消息发出去后执行，保证事务的一致。
     *
     * @param msg
     * @param arg
     * @return
     */
    @Override
    @Transactional
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        Account account = JSON.parseObject(msg.getBody(), Account.class);
        account.setTransactionId(msg.getTransactionId());
        //设置该消息
        LocalTransactionState state = LocalTransactionState.ROLLBACK_MESSAGE;
        try {
            boolean isCommit = accountService.saveOrUpdate(account);
            if (isCommit) {
                state = LocalTransactionState.COMMIT_MESSAGE;
            } else {
                state = LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            log.error("执行本地事务异常：{}", e);
            throw  new RuntimeException(e);
        }
        log.info("执行本地事务成功");
        return state;
    }

    /**
     * 检查本地事务执行状态
     * @param transactionId
     * @return
     */
    public boolean checkTransferStatus(String transactionId) {
        //根据本地线程Id是否有这个数据，有运单记录，标识本地事务执行成功
        int count = accountService.count(Wrappers.<Account>lambdaQuery().eq(Account::getTransactionId, transactionId));
        return count > 0;
    }

    /**
     * 每隔一段时间 rocketMQ 会回调 这个方法 判断 每一条消息是否提交。防止 消息状态停滞 或者出现超时的情况
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        LocalTransactionState state = LocalTransactionState.UNKNOW;
        try {
            Account account = JSON.parseObject(msg.getBody(), Account.class);
            if (account != null) {
                state = LocalTransactionState.ROLLBACK_MESSAGE;
            }
            log.info("消费端调用" + msg.getTransactionId());
            boolean isCommit = checkTransferStatus(msg.getTransactionId());
            if (isCommit){
                state = LocalTransactionState.COMMIT_MESSAGE;
            }else {
                state = LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            log.error("checkLocalTransaction异常：{}", e);
            throw new RuntimeException(e);
        }
        return state;
    }
}
